#include <stdio.h>
#include <pcap.h>
#include <unistd.h>
#include <stdlib.h>
#include <string.h>
#include <getopt.h>
#include <errno.h>
#include <sys/wait.h>
#include <sys/resource.h>
#include <netinet/in.h>
#include <netinet/tcp.h>

#include "redisoper.h"
#include "global.h"
#include "logger.h"

#define MAXOF(t) (((t)(~((t)0))) > ((t)0) ? ((t)(~((t)0))) : ((t)((~((t)0))^(((t)1)<<((sizeof(t)<<3)-1)))))

#define PROMISC_ENABLE              1
#define PROMISE_DISABLE             0
#define PCAP_ERRBUF_SIZE	        256
#define MAX_PACKET_LEN              1600
#define PACKET_BUFF_SIZE            1600
#define PACKET_TYPE_DIRECT          0x80

#define uChar 						unsigned char
#define uLong 						unsigned long
#define uIn16 						unsigned short
#define uIn32 						unsigned int
#define uIn64                       uint64_t


static char*	             s_pEtherName;
static pthread_t             s_pThreadSec;
static pcap_t*               s_pCapObject;
static volatile uIn64        s_nIncBytes;
static volatile uIn64        s_nTotalBytes;
static volatile uIn64        s_nTotalPackage;


int GetExeAbsPath(char* pExeAbsPath, size_t zLen)
{
    char  szProcPidExe[300] = { 0 };
    char  szExeAbsPath[300] = { 0 };

    ssize_t z = 0;
    char*   p = NULL;

    sprintf(szProcPidExe, "/proc/%d/exe", getpid());
    z = readlink(szProcPidExe, szExeAbsPath, sizeof(szExeAbsPath));
    if(z > 0)
    {
        p = strrchr(szExeAbsPath, '/');
        if(!p)
            return -1;

        z = p - szExeAbsPath;
        if(z > zLen - 1)
            return -2;

        strncpy(pExeAbsPath, szExeAbsPath, z);
        *(pExeAbsPath + z) = '\0';
        return 0;
    }

    return -3;
}


pcap_t* libpcap_open_and_prepare(char* szName, int nBufSize)
{
    int         i, bufsize;
    char        szCapbuf[256];
    char        szErrbuf[PCAP_ERRBUF_SIZE];
    socklen_t   nBufsizeLen = sizeof(bufsize);
    pcap_t*     pCapObj = NULL ;

    struct bpf_program filter;
    char* filter_app;

	if(g_htProfile.m_bMqttCommand)
	{
		filter_app = g_htProfile.m_szMqttFilterRule;
	}
	else if(g_htProfile.m_bRestCommand)
	{
		filter_app = g_htProfile.m_szRestFilterRule;
	}
	else if(g_htProfile.m_bTzCommand)
	{
		filter_app = g_htProfile.m_szTzFilterRule;
	}
	else
	{
		exit(1);
	}

    memset(szErrbuf,0,PCAP_ERRBUF_SIZE);
    memset(szCapbuf,0,256);
    strcpy(szCapbuf,szName);

    pCapObj=pcap_open_live(szCapbuf,MAX_PACKET_LEN,PROMISC_ENABLE,1000,szErrbuf);
    if(!pCapObj)
    {
		LOG__ERROR("libpcap open live error :%s",szErrbuf);
        exit(1);
    }

    pcap_compile(pCapObj, &filter, filter_app, 0, 0);
    pcap_setfilter(pCapObj, &filter);

    i=pcap_datalink(pCapObj);

    if(i!=DLT_EN10MB&&i!=DLT_IEEE802)
    {
		LOG__WARN("Datalink layer not to be supported! %d",i);
        pcap_close(pCapObj);
        exit(1);
    }


    //set bufsize
    bufsize = nBufSize;
    if(setsockopt(((int*)pCapObj)[0], SOL_SOCKET, SO_RCVBUF, (char *)&bufsize, sizeof(bufsize)))
    {
        LOG__INFO("set pcap buff error size=%d",bufsize);
    }
    else
    {
        LOG__INFO("set pcap buff size=%d",bufsize);
    }

    //get bufsize
    if(getsockopt(((int*)pCapObj)[0], SOL_SOCKET, SO_RCVBUF, (char *)&bufsize, &nBufsizeLen))
    {
        LOG__INFO("get pcap buff size failor");
    }
    else
    {
        LOG__INFO("get pcap buff size=%d",bufsize);
    }

    return pCapObj;
}


void  libpcap_loop_proc(u_char * usrname __attribute__((__unused__)),const struct pcap_pkthdr* h,const u_char * data)
{
   __sync_add_and_fetch(&s_nIncBytes,h->caplen);
   __sync_add_and_fetch(&s_nTotalBytes,h->caplen);
   __sync_add_and_fetch(&s_nTotalPackage,1);
    return;
}


void libpcap_close(pcap_t* obj)
{
    if(obj != NULL )
    {
        pcap_close(obj);
        obj = NULL ;
    }
}

void doEthCapture()
{
    int rst;
    //initilize pcap object
    s_pCapObject=libpcap_open_and_prepare(s_pEtherName, 12000000);
    if(!s_pCapObject) return;

    pcap_setnonblock(s_pCapObject,1,NULL);//nonblock mode

    //loop
    LOG__INFO("start pcap capturing");
    while(!g_nExitFlag)
    {
        rst = pcap_dispatch(s_pCapObject,100,libpcap_loop_proc, NULL);
        if(rst < 0)
        {
	    	LOG__ERROR("pcap_loop error, exist");
            g_nExitFlag = 1;
        }
        else if(rst == 0)
        {
            usleep(1000);
        }
    }

    libpcap_close(s_pCapObject);
    LOG__INFO("close pcap handle");
}

void redisCommit(int bFirst)
{
	char				szCommand[128]         = { 0 };
	char				szCommandDaily[128]    = { 0 };
	char				szCommandExp[128]      = { 0 };
	char				szDate[20]	           = { 0 };
	time_t		        t				       = 0;
	struct tm	        tm				       = { 0 };

	t = time(NULL);
	localtime_r(&t, &tm);
	sprintf(szDate, "%d-%02d-%02d", 1900+tm.tm_year, 1+tm.tm_mon, tm.tm_mday);

	if(g_htProfile.m_bMqttCommand)
	{
		sprintf(szCommand,"INCRBY htstream:collection:captured:mqtt %lu",s_nIncBytes);
		sprintf(szCommandDaily,"INCRBY htstream:collection:captured:%s:mqtt %lu",szDate,s_nIncBytes);
	}
	else if(g_htProfile.m_bRestCommand)
	{
		sprintf(szCommand,"INCRBY htstream:collection:captured:rest %lu",s_nIncBytes);
		sprintf(szCommandDaily,"INCRBY htstream:collection:captured:%s:rest %lu",szDate,s_nIncBytes);
	}
	else if(g_htProfile.m_bTzCommand)
	{
		sprintf(szCommand,"INCRBY htstream:collection:captured:tz %lu",s_nIncBytes);
		sprintf(szCommandDaily,"INCRBY htstream:collection:captured:%s:tz %lu",szDate,s_nIncBytes);
	}
	redisSet(szCommand);
	redisSet(szCommandDaily);

	if((tm.tm_hour==0 && tm.tm_min==0 && tm.tm_sec==0) || (bFirst==1))
	{
		if(g_htProfile.m_bMqttCommand)
		{
			sprintf(szCommandExp,"EXPIRE htstream:collection:captured:%s:mqtt %u",szDate,10*24*3600);
		}
		else if(g_htProfile.m_bRestCommand)
		{
			sprintf(szCommandExp,"EXPIRE htstream:collection:captured:%s:rest %u",szDate,10*24*3600);
		}
		else if(g_htProfile.m_bTzCommand)
		{
			sprintf(szCommandExp,"EXPIRE htstream:collection:captured:%s:tz %u",szDate,10*24*3600);
		}

	    redisSet(szCommandExp);
	}

	__sync_and_and_fetch(&s_nIncBytes, 0);

}



static void* RedisThreadSec(void* pRedisCon)
{
	time_t			tCurr 			= 0;
	time_t			tLast 			= 0;
    int             bFirst          = 1;
    while (1){

		if(g_nExitFlag)
		   break;

	   tCurr = time(NULL);

	   if(0==(tCurr%5) && tCurr!=tLast)
	   {
	   		redisCommit(bFirst);
			LOG__INFO("----------captured %lu packages %lu bytes----------",s_nTotalPackage,s_nTotalBytes);
			tLast = tCurr;
            bFirst = 0;
	   }
       else
       {
            sleep(1);
            continue;
       }

    }
    return NULL;
}



int InitCapture()
{
	pthread_create(&s_pThreadSec, NULL, RedisThreadSec, NULL);
	return 0;
}

void DeinitCapture()
{
    pthread_join(s_pThreadSec, NULL);
}

void run()
{
	if(0==InitLogger()     &&
	   0==InitZkProfile()  &&
	   0==MyRedisConnect() &&
	   0==InitCapture())
	{
		doEthCapture();

		DeinitRedis();
		DeinitCapture();
		DeinitLogger();
	}
	return;

}

void usage(const char* pCmdName)
{
    printf("usage: %s --i=ethname --mqtt|--rest|--tz --zookeeper=ip:port\n"
           "       %s -i <ethname> {-m | -r | -t} -z <ip:port>\n",
           pCmdName, pCmdName);
}

int main(int argc,char** argv)
{
    int         nDbgMod  = 0;

    pid_t       pid      = 0;
    int         i        = 0;
    int         c        = 0;
	int			m		 = 0;
    int         status   = 0;
    const char* pCmdName = NULL;

    struct option long_options[] =
    {
        {"debug",       no_argument,       NULL, 'd'},
        {"tz",          no_argument,       NULL, 't'},
        {"mqtt",        no_argument,       NULL, 'm'},
        {"rest",        no_argument,       NULL, 'r'},
        {"zookeeper",   required_argument, NULL, 'z'},
        {"i",           required_argument, NULL, 'i'},
        {"help",        no_argument,       NULL, 'h'},
        {NULL,          0,                 NULL,  0 },
    };

    struct rlimit rlmt = {MAXOF(rlim_t), MAXOF(rlim_t)};
    if(setrlimit(RLIMIT_CORE, &rlmt) == -1)
    {
        printf("set core file size failed, errno=%d", errno);
        return -1;
    }


    pCmdName = strrchr(argv[0], '/');
    if(pCmdName)
        ++pCmdName;
    else
        pCmdName = argv[0];


    while(1)
    {
        c = getopt_long(argc, argv, "+:dtmrz:i:h", long_options, &i);

        if(-1 == c)
            break;

        switch(c)
        {
            case 'd':
                nDbgMod = 1;
				++m;
                break;

	        case 't':
		        g_htProfile.m_bTzCommand = 1;
				++m;
		    break;

            case 'm':
                g_htProfile.m_bMqttCommand = 1;
				++m;
                break;

            case 'r':
                g_htProfile.m_bRestCommand = 1;
				++m;
                break;

            case 'z':
                g_pZookeeper = optarg;
				++m;
                break;

	        case 'i':
		        s_pEtherName = optarg;
				++m;
		        break;

            case 'h':
                usage(pCmdName);
                return 0;

            default:
                break;
        }
    }

    if(m != 3)
    {
        usage(pCmdName);
        return -1;
    }

    if((g_htProfile.m_bTzCommand+g_htProfile.m_bMqttCommand+g_htProfile.m_bRestCommand)!=1 || !g_pZookeeper)
    {
        usage(pCmdName);
        return -1;
    }

    strcpy(g_szInstName, pCmdName);
    GetExeAbsPath(g_szInstPath, sizeof(g_szInstPath));
    strcpy(g_szConfPath, g_szInstPath);
    strcat(g_szConfPath, "/conf");

    if(nDbgMod)
    {
        run();
        return 0;
    }

    if(daemon(1, 0) < 0)
    {
        perror("omg! daemon returns error ...\n");
        return -1;
    }

    while(1)
    {
        pid = fork();
        if(pid  <  0)
            continue;

        if(pid == 0)
        {
            run();
        }
        else if(pid > 0)
        {
            while(1)
            {
                if(wait(&status)>0)
                    break;
            }
        }
    }

    return 0;

}

